package hbec.kettle.etl.plugins.writer.mysqlwriter;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import hbec.kettle.etl.common.constants.DataBase;
import hbec.kettle.etl.common.exception.ExceptionTracker;
import hbec.kettle.etl.common.plugin.Result;
import hbec.kettle.etl.common.plugin.Writer;
import hbec.kettle.etl.engine.conf.TransEntityConf;
import hbec.kettle.etl.engine.plugin.Line;
import hbec.kettle.etl.plugins.common.Cache;
import hbec.kettle.etl.plugins.core.TransExecutor;

public class MysqlWriter extends Writer implements Callable<Result> {
	private static final Logger logger = LoggerFactory.getLogger(MysqlWriter.class);
	
	private TransEntityConf transEntityConf;
	Map<DataBase, String> watchSumTablesOfDb;
	private Line line;
	private CountDownLatch latch;
	
	public MysqlWriter(Map<DataBase, String> watchSumTablesOfDb, 
			TransEntityConf transEntityConf, Line line, CountDownLatch latch) {
		this.transEntityConf = transEntityConf;
		this.watchSumTablesOfDb = watchSumTablesOfDb;
		this.line = line;
		this.latch = latch;
	}

	@Override
	public Result call() throws Exception {
		Result res = new Result();
		try {
			TransExecutor transExecutor = Cache.getTransExecutorCache(transEntityConf);
			Map<String, String> variables = new HashMap<>();
			variables.put("MAX_POS", String.valueOf(line.getMaxPosition()));
			variables.put("MIN_POS", String.valueOf(line.getMinPosition()));
			variables.put("WATCH_SUMTABLE", watchSumTablesOfDb.get(DataBase.valueOf(line.getObject().split("\\.")[0].trim().toUpperCase())));
			variables.put("WATCH_TABLE", line.getObject().split("\\.")[1].trim());
			transExecutor.setVariables(variables);
			res = transExecutor.execute();
		} catch (Exception e) {
			logger.error(ExceptionTracker.trace(e));
		} finally {
			latch.countDown();
		}
		return res;
	}
	
}
